package fun.tan90.device.source.mqtt.plugin.config;

import cn.hutool.core.util.RandomUtil;
import fun.tan90.device.event.model.DeviceSourceData;
import fun.tan90.device.source.core.SourceFactory;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.core.client.IMqttClientConnectListener;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.dromara.mica.mqtt.core.client.MqttClientSubscription;
import org.noear.snack.ONode;
import org.noear.solon.annotation.Component;
import org.tio.core.ChannelContext;

import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

@Slf4j
@Component
public class MqttClientFactory extends SourceFactory {

    @Override
    public Class<?> init(ONode connectNode, BiConsumer<String, Object> biConsumer) {
        for (ONode node : connectNode.ary()) {
            String name = node.get("name").getString();
            ONode connect = node.get("connect");
            MqttClientTemplate mqttClientTemplate = new MqttClientTemplate(MqttClient.create()
                    .ip(connect.get("ip").getString())
                    .port(connect.get("port").getInt())
                    .username(connect.get("username").getString())
                    .password(connect.get("password").getString())
                    .clientId(connect.get("clientIdPrefix").getString() + RandomUtil.randomNumbers(6))
                    .keepAliveSecs(connect.get("keepAliveSecs").getInt())
                    .retryCount(connect.get("retryCount").getInt())
                    .connectListener(new IMqttClientConnectListener() {
                        @Override
                        public void onConnected(ChannelContext context, boolean isReconnect) {
                            if (isReconnect) {
                                log.warn("{} 重连 mqtt 服务器重连成功...", name);
                            } else {
                                log.info("{} 连接 mqtt 服务器成功...", name);
                            }
                            clientInitSuccess(name);
                        }

                        @Override
                        public void onDisconnect(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) {
                            log.error("{} mqtt 链接断开 remark:{} isRemove:{}", name, remark, isRemove, throwable);
                            clientInitFailure(name, throwable);
                        }
                    })
                    .connect());
            biConsumer.accept(name, mqttClientTemplate);
        }
        return MqttClientTemplate.class;
    }

    @Override
    public void post(Map<String, ONode> ruleMap, Map<String, ?> clientMap) {
        for (String name : clientMap.keySet()) {
            ONode node = ruleMap.get(name).get("topic");
            MqttClientTemplate mqttClientTemplate = (MqttClientTemplate) clientMap.get(name);
            List<MqttClientSubscription> mqttClientSubscriptionList = node.ary().stream().map(n -> {
                        MqttQoS qos = MqttQoS.valueOf(n.get("qos").getInt());
                        String topicPattern = n.get("topic-pattern").getString();
                        int index = n.get("index").getInt();
                        return new MqttClientSubscription(qos, topicPattern, (context, topic, message, payload) -> {
                            log.info("name:{},topic:{},message:{}", name, topic, message);
                            afterPost(new DeviceSourceData(topicPattern, topic.split("/")[index], new String(payload)));
                        });
                    }
            ).collect(Collectors.toList());
            mqttClientTemplate.subscribe(mqttClientSubscriptionList);
        }
    }
}
